package com.example.rpcdemo.core.client;

import com.example.rpcdemo.core.protocol.RpcResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.concurrent.ConcurrentHashMap;

public class RpcClientHandler extends ChannelInboundHandlerAdapter {
    private final ConcurrentHashMap<String, RpcFuture> pendingRequests;
    
    public RpcClientHandler(ConcurrentHashMap<String, RpcFuture> pendingRequests) {
        this.pendingRequests = pendingRequests;
    }
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        RpcResponse response = (RpcResponse) msg;
        String requestId = "";
        
        for (String key : pendingRequests.keySet()) {
            if (pendingRequests.get(key) != null) {
                requestId = key;
                break;
            }
        }
        
        if (!requestId.isEmpty()) {
            RpcFuture future = pendingRequests.remove(requestId);
            if (future != null) {
                if (response.hasError()) {
                    future.setFailure(new Exception(response.getError()));
                } else {
                    future.setSuccess(response.getResult());
                }
            }
        }
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}